package com.youxin.infra.util;

import com.youxin.infra.kafkaUtil.KafkaProperties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProduUtil {

    private final KafkaProducer<Integer, String> producer;
    private final String topic;

    public KafkaProduUtil() {
        Properties props = new Properties();
        props.put("bootstrap.servers", KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
        props.put("client.id", KafkaProperties.CLIENT_ID);
        props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer<>(props);
        this.topic = KafkaProperties.TOPIC_SINK;
    }

    public static KafkaProduUtil getInstance() {
        return Singleton.instance;
    }

    private static class Singleton {
        private static KafkaProduUtil instance = new KafkaProduUtil();
    }

    public void sendMess(String message) {
        producer.send(new ProducerRecord(topic, message), new DemoCallBack(System.currentTimeMillis(), message));
        System.out.println("Sent message: (" + ", " + message + ")");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
